package cas.ihep.hmss.hdfsin;

/**
 * Created by chocolate on 2017/11/24.
 */

import cas.ihep.util.RefreshFileSize;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hdfs.DFSClient;
import org.apache.hadoop.hdfs.protocol.HdfsFileStatus;
import org.apache.hadoop.hdfs.protocol.LocatedBlock;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

import java.io.File;
import java.io.IOException;
import java.lang.reflect.Constructor;
import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
import java.net.URI;
import java.net.URISyntaxException;

public class InMapper extends Mapper<LongWritable, Text, Text, Text> {
    DFSClient dfs = null;
    Object s4h = null;
    static final Class<?> Service4HepClass;
    static final Constructor<?> constructor;
    static final Method createFile,addBlock,getTmpFile,calculatefileCheckSums,close0,close1;
    static {
        try {
            Service4HepClass=Class.forName("org.apache.hadoop.hdfs.Service4Hep");
            constructor=Service4HepClass.getConstructor(DFSClient.class,Configuration.class);
            constructor.setAccessible(true);
            createFile=Service4HepClass.getMethod("createFile",String.class,boolean.class,short.class,long.class);
            createFile.setAccessible(true);
            addBlock=Service4HepClass.getMethod("addBlock",String.class,long.class);
            addBlock.setAccessible(true);
            getTmpFile=Service4HepClass.getMethod("getTmpFile",LocatedBlock.class);
            getTmpFile.setAccessible(true);
            calculatefileCheckSums=Service4HepClass.getMethod("calculatefileCheckSums",String.class,LocatedBlock.class);
            calculatefileCheckSums.setAccessible(true);
            close0=Service4HepClass.getMethod("close",LocatedBlock.class,String.class,long.class);
            close0.setAccessible(true);
            close1=Service4HepClass.getMethod("close");
            close1.setAccessible(true);
        }catch (RuntimeException e){
            e.printStackTrace();
            throw e;
        }catch (Exception e){
            e.printStackTrace();
            throw new RuntimeException(e);
        }

    }
    //    public void run(Context context) throws IOException, InterruptedException { /*context上下文对象，存储一些job conf的信息，*/
//        setup(context);
//        JobID jobId = context.getJobID();
//        FileSplit split = (FileSplit)context.getInputSplit();
//        Path path = split.getPath();
//        map(jobId, path, context);
//        cleanup(context);
//    }
    protected void map(LongWritable lno, Text line, Context context) throws IOException,
            InterruptedException {
        //Path paths[] = {path};
        try {
            String attemp = context.getTaskAttemptID().toString();
            LocatedBlock lb = null;
            HdfsFileStatus hfstatus = null;
            String opath = "";
            //boolean isWrite2hdfs = true;
            //boolean exception = false;
            RefreshFileSize rfs = null;
            String path = line.toString();/*输入文件里的每一行*/
            Configuration conf = context.getConfiguration();
            String nn = conf.get("fs.default.name", "hdfs://192.168.60.41:8020");/*namenode*/
            String targetdir = conf.get("target.directory");
            URI uri;
            try {
                uri = new URI(nn);

            } catch (URISyntaxException e) {
                throw new IllegalArgumentException(e);
            }
            dfs = new DFSClient(uri, conf);
            //s4h = new Service4Hep(dfs,conf);
            s4h = constructor.newInstance(dfs, conf);
            rfs = new RefreshFileSize(dfs);
            int lastSlash = path.lastIndexOf('/');/*获取文件名*/
            String dest = targetdir + '/' + path.substring(lastSlash + 1);/*完整的目标逻辑路径*/
            //hfstatus = s4h.createFile(dest,true,(short) 1,2147483648L);/*新建文件获取元数据信息*/
            hfstatus = (HdfsFileStatus) createFile.invoke(s4h, dest, true, (short) 1, 2147483648L);
            //lb = s4h.addBlock(dest, hfstatus.getFileId());/*为文件分配block*/
            lb=(LocatedBlock) addBlock.invoke(s4h,dest,hfstatus.getFileId());
            //String tmp = s4h.getTmpFile(lb);
            String tmp=(String)getTmpFile.invoke(s4h,lb);
            opath = tmp + "/" + lb.getBlock().getBlockName();/*实际的物理路径*/
            rfs.put(dest, hfstatus.getFileId(), opath);
            ProcessBuilder builder = new ProcessBuilder();
            builder.command("time", "globus-url-copy", "-vb", "-p", "3", "ftp://xingg:123456@helion02.ihep.ac.cn:5555/" + path, opath);
            //File tmpdir=new File("/tmp");
            //builder.redirectError(File.createTempFile("hmsjob",".err",tmpdir));
            //builder.redirectOutput(File.createTempFile("hmsjob",".out",tmpdir));
            builder.redirectOutput(ProcessBuilder.Redirect.INHERIT);
            builder.redirectError(ProcessBuilder.Redirect.INHERIT);
            Process process = builder.start();
            int exitValue = process.waitFor();//等待process结束
            if (exitValue == 0) {
                rfs.remove(dest);
                File f = new File(opath);
                if (f.exists()) {
                    //s4h.calculatefileCheckSums(opath, lb);
                    calculatefileCheckSums.invoke(s4h,opath,lb);
                    lb.getBlock().setNumBytes(f.length());
                    //s4h.close(lb, dest, hfstatus.getFileId());
                    close0.invoke(s4h,lb,dest,hfstatus.getFileId());
                } else {
                    dfs.delete(dest, true);
                }
                rfs.close();
                //s4h.close();
                close1.invoke(s4h);
                dfs.close();
                return;
            }
            System.err.println("Error");
        }catch (InvocationTargetException | InstantiationException | IllegalAccessException e){
            throw new IOException(e);
        }
    }

}
